package edu.xhu.search.service.impl;

import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONUtil;
import edu.xhu.api.user.UserClient;
import edu.xhu.model.biliuser.dtos.UserInfosDto;
import edu.xhu.model.common.dtos.ResponseResult;
import edu.xhu.model.common.enums.AppHttpCodeEnum;
import edu.xhu.model.search.constant.SearchConstant;
import edu.xhu.model.search.pojos.VideoSearchDoc;
import edu.xhu.model.search.vos.HotSearchVo;
import edu.xhu.model.search.vos.VideoSearchVo;
import edu.xhu.search.service.SearchService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.*;

import org.springframework.data.redis.core.ZSetOperations.TypedTuple;

@Service
@RequiredArgsConstructor
@Slf4j
public class SearchServiceImpl implements SearchService {

    private final UserClient userClient;

    private final RestHighLevelClient restHighLevelClient;

    private final KafkaTemplate<String, String> kafkaTemplate;

    private final StringRedisTemplate redisTemplate;

    /**
     * 根据搜索条件搜索视频
     *
     * @param keyword
     * @param pageNo
     * @param pageSize
     * @return
     */
    @Override
    public ResponseResult<?> searchVideosByKeyword(String keyword, Integer pageNo, Integer pageSize) {
        List<VideoSearchVo> searchVideoResult; //最终结果

        // 1. 先走 ES 做分页查询
        // 1.1 创建Request，组织请求参数
        SearchRequest request = new SearchRequest(SearchConstant.VIDEO_INDEX_NAME);
        // 搜索条件参数
        request.source().query(QueryBuilders.matchQuery("videoTitle", keyword));
        // 分页参数
        request.source().from((pageNo - 1) * pageSize).size(pageSize);
        // 高亮条件
        request.source().highlighter(
                SearchSourceBuilder.highlight()
                        .field("videoTitle")
                        .preTags("<em>")
                        .postTags("</em>")
        );


        try {
            // 1.2 发送请求
            SearchResponse searchResponse = restHighLevelClient.search(request, RequestOptions.DEFAULT);

            //1.3 解析 ES 响应结果
            searchVideoResult = handleVideoSearchResponse(searchResponse);
        } catch (IOException e) {
            return ResponseResult.errorResult(AppHttpCodeEnum.SERVER_ERROR);
        }


        //最后的判空逻辑
        if (searchVideoResult == null)
            return ResponseResult.errorResult(AppHttpCodeEnum.SERVER_ERROR);

        // 2. 发送消息刷新热搜词条
        kafkaTemplate.send(SearchConstant.HOT_SEARCH_PROCESSING_TOPIC, keyword);

        return ResponseResult.okResult(searchVideoResult);
    }

    /**
     * @return 热搜列表
     */
    @Override
    public ResponseResult getHotSearch() {

        List<HotSearchVo> hotSearchVoList = new ArrayList<>(); //最终结果集

        // 1. 获取分数最高的前 10 个热搜及其分数（热度）
        Set<TypedTuple<String>> topElements = redisTemplate.opsForZSet()
                .reverseRangeWithScores(SearchConstant.HOT_SEARCH_RANKING_ZSET, 0, 9);

        //判空逻辑
        if (topElements == null || topElements.size() == 0)
            return ResponseResult.okResult(null);

        for (TypedTuple<String> element : topElements) {
            HotSearchVo hotSearchVo = new HotSearchVo();
            hotSearchVo.setContent(element.getValue()).
                    setScore(Double.parseDouble(element.getScore().toString()));
            hotSearchVoList.add(hotSearchVo);
        }

        // 2. 查询这些热搜的更新时间，并将更新时间装入 hotSearchVoList 结果集
        List<Object> timeResults = redisTemplate.executePipelined((RedisCallback<?>) (connection) -> {
            for (HotSearchVo hotSearchVo : hotSearchVoList) {
                connection.hGet(
                        redisTemplate.getStringSerializer().serialize(SearchConstant.HOT_SEARCH_UPDATE_TIME_HSET),
                        redisTemplate.getStringSerializer().serialize(hotSearchVo.getContent())
                );
            }
            return null;
        });

        for (int i = 0; i < timeResults.size(); i++) {
            Long timestamp = timeResults.get(i) != null ? Long.parseLong(timeResults.get(i).toString()) : 0L;
            hotSearchVoList.get(i).setUpdateTimeStamp(timestamp);
        }

        // 3. 判断热搜的类型（更新时间，热度已经装配好）
        // 3.1 按 updateTimeStamp 字段降序排序，设置前一半元素为 ”新“ 热搜
        hotSearchVoList.sort(Comparator.comparing(HotSearchVo::getUpdateTimeStamp).reversed());

        for (int i = 0; i < hotSearchVoList.size(); i++) {
            if (i > hotSearchVoList.size() / 4)
                break;
            hotSearchVoList.get(i).setType(1);
        }

        // 3.2 按 score 字段降序排序，设置前一半元素为 ”热“ 热搜
        hotSearchVoList.sort(Comparator.comparing(HotSearchVo::getScore).reversed());

        for (int i = 0; i < hotSearchVoList.size(); i++) {
            if (i > hotSearchVoList.size() / 4)
                break;
            hotSearchVoList.get(i).setType(2);
        }

        return ResponseResult.okResult(hotSearchVoList);
    }

    /**
     * 解析视频搜索 ES 响应
     *
     * @param response
     * @return 包装好的返回结果 List<VideoSearchVo>
     */
    private List<VideoSearchVo> handleVideoSearchResponse(SearchResponse response) {

        List<VideoSearchVo> videoSearchVoList = new ArrayList<>(); //最终结果集

        ResponseResult<?> responseResult = null; //用来保存feign之调用结果
        List<Map> userInfoMapList = null; //用户全部信息之map列表

        // 1. 遍历 ES 结果数组
        SearchHits searchHits = response.getHits();

        SearchHit[] hits = searchHits.getHits();
        for (SearchHit hit : hits) {
            // 1.1 得到_source，也就是原始json文档
            String source = hit.getSourceAsString();

            //反序列化，拿到文档
            VideoSearchDoc videoSearchDoc = JSONUtil.toBean(source, VideoSearchDoc.class);

            //拿到文档对应的 VideoSearchVo
            VideoSearchVo videoSearchVo = JSONUtil.toBean(videoSearchDoc.getVideoJson(), VideoSearchVo.class);

            // 1.2 得到每个文档的算分
            Double score = (double) hit.getScore();
            videoSearchVo.setScore(score);

            // 1.3 获取高亮结果
            Map<String, HighlightField> hfs = hit.getHighlightFields();

            if (hfs != null && !hfs.isEmpty()) {
                //有高亮结果，获取 videoTitle 的高亮结果
                HighlightField hf = hfs.get("videoTitle");
                if (hf != null) {
                    // 获取第一个高亮结果片段，就是视频标题的高亮值
                    String hfVideoTitle = hf.getFragments()[0].string();
                    videoSearchVo.setHighlightedTitle(hfVideoTitle);
                }
            }


            // 1.4 添加 videoSearchVo 至结果集
            videoSearchVoList.add(videoSearchVo);
        }

        //判空逻辑
        if (videoSearchVoList == null || videoSearchVoList.size() == 0)
            return videoSearchVoList;

        // 2. 开始组装用户数据至结果集
        // 2.1 拿到所涉及到的全部用户 id
        Set<Long> userIds = new HashSet<>();
        videoSearchVoList.forEach(videoSearchVo -> {
            userIds.add(videoSearchVo.getBiliVideo().getUserId());
        });
        UserInfosDto userInfosDto = new UserInfosDto();
        userInfosDto.setUserIds(new ArrayList<Long>(userIds));

        // 2.2 远程调用 user 微服务，去拿用户信息
        responseResult = userClient.queryAllUserInfoByIds(userInfosDto);
        if (!responseResult.getCode().equals(200)) {
            return null;
        }

        JSONArray userInfoJsonArray = JSONUtil.parseArray(responseResult.getData());
        userInfoMapList = JSONUtil.toList(userInfoJsonArray, Map.class);

        // 2.3 组装用户数据至结果集
        // 用户信息查找表，{userId=infoMap,userId=infoMap ,..., userId=infoMap,}
        Map<Long, Map> userIdMapToInfo = new HashMap<>();
        userInfoMapList.forEach(userInfoMap -> {
            userIdMapToInfo.put(Long.parseLong(userInfoMap.get("user_id").toString()), userInfoMap);
        });

        videoSearchVoList.forEach(videoSearchVo -> {
            videoSearchVo.setUserAllInfo(userIdMapToInfo.get(videoSearchVo.getBiliVideo().getUserId()));
        });

        return videoSearchVoList;
    }
}
